package com.study;

import org.apache.commons.compress.utils.Lists;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * @author sjw
 * @description
 * @date 2019-08-19 上午11:43
 */
public class SourceRabbitMqSinkPatchMySQL {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        //这些配置建议可以放在配置文件中，然后通过 parameterTool 来获取对应的参数值
        final RMQConnectionConfig connectionConfig = new RMQConnectionConfig
                .Builder().setHost("localhost").setVirtualHost("/")
                .setPort(5672).setUserName("root").setPassword("123456")
                .build();

        SingleOutputStreamOperator<User> dataStreamSource = env.addSource(new RMQSource<>(connectionConfig,
                "q.study.flink",
                true,
                new SimpleStringSchema()))
                .setParallelism(1).map(new MapFunction<String, User>() {
                    @Override
                    public User map(String s) throws Exception {
                        return JsonHelper.jsonToObject(s, User.class);
                    }
                });

        //env.addSource()
        dataStreamSource.timeWindowAll(Time.minutes(1)).apply(new AllWindowFunction<User, List<User>, TimeWindow>() {
            @Override
            public void apply(TimeWindow timeWindow, Iterable<User> iterable, Collector<List<User>> collector) throws Exception {
                List<User> users= new ArrayList<>();
                iterable.forEach(users::add);
                if(users.size()>0){
                    System.out.println("1 分钟内收集到 student 的数据条数是：" + users.size());
                    collector.collect(users);
                }
            }
        }).addSink(new SinkPatchMySQL());
        //dataStreamSource.addSink(new SinkMySQL());


        //如果想保证 exactly-once 或 at-least-once 需要把 checkpoint 开启
//        env.enableCheckpointing(10000);
        env.execute("flink learning connectors rabbitmq");

    }
}
